/*
 * reorderbuffer.h
 *	  PostgreSQL 逻辑重放/重排序缓冲区管理。
 *
 * Copyright (c) 2012-2022, PostgreSQL Global Development Group
 *
 * src/include/replication/reorderbuffer.h
 */
#ifndef REORDERBUFFER_H
#define REORDERBUFFER_H

#include "access/htup_details.h"
#include "lib/ilist.h"
#include "storage/sinval.h"
#include "utils/hsearch.h"
#include "utils/relcache.h"
#include "utils/snapshot.h"
#include "utils/timestamp.h"

extern PGDLLIMPORT int logical_decoding_work_mem;

/* 单个元组，存储在一块内存中 */
typedef struct ReorderBufferTupleBuf
{
	/* 在预分配列表中的位置 */
	slist_node	node;

	/* 元组头，对于逻辑解码的用户来说，最有趣的部分 */
	HeapTupleData tuple;

	/* 元组缓冲区的预分配大小，不同于元组大小 */
	Size		alloc_tuple_size;

	/* 实际的元组数据 follows */
} ReorderBufferTupleBuf;

/* 指向存储在 TupleBuf 中的数据的指针 */
#define ReorderBufferTupleBufData(p) \
	((HeapTupleHeader) MAXALIGN(((char *) p) + sizeof(ReorderBufferTupleBuf)))

/*
 * 传递给 'change' 回调的变更类型。
 *
 * 出于效率和简单性的考虑，我们希望将快照、命令ID
 * 和组合CID与用户可见的 INSERT/UPDATE/DELETE 变更放在同一列表中。
 * 解码设施的用户永远不会看到带有 *_INTERNAL_* 操作的变更。
 *
 * INTERNAL_SPEC_INSERT、INTERNAL_SPEC_CONFIRM 和 INTERNAL_SPEC_ABORT
 * 变更涉及“投机性插入”、其确认和相应的中止。它们用于 INSERT .. ON CONFLICT .. UPDATE。
 * 逻辑解码的用户无需关心这些。
 */
typedef enum ReorderBufferChangeType
{
	REORDER_BUFFER_CHANGE_INSERT,
	REORDER_BUFFER_CHANGE_UPDATE,
	REORDER_BUFFER_CHANGE_DELETE,
	REORDER_BUFFER_CHANGE_MESSAGE,
	REORDER_BUFFER_CHANGE_INVALIDATION,
	REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT,
	REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID,
	REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID,
	REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT,
	REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM,
	REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT,
	REORDER_BUFFER_CHANGE_TRUNCATE
} ReorderBufferChangeType;

/* 前向声明 */
struct ReorderBufferTXN;

/*
 * 单个的 'change'，可以是插入（带一个元组）、更新（旧的，新），
 * 或删除（旧）。
 *
 * 同样的结构也在内部用于其他目的，但这在 reorderbuffer.c 
 * 外部永远不应可见。
 */
typedef struct ReorderBufferChange
{
	XLogRecPtr	lsn;

	/* 变更的类型。 */
	ReorderBufferChangeType action;

	/* 此变更所属的事务。 */
	struct ReorderBufferTXN *txn;

	RepOriginId origin_id;

	/*
	 * 变更的上下文数据。联合体中哪个部分有效取决于
	 * 操作。
	 */
	union
	{
		/* 当操作 == *_INSERT|UPDATE|DELETE 时的旧元组和新元组 */
		struct
		{
			/* 已经更改的关系 */
			RelFileNode relnode;

			/* 不再需要以前重新组装的 toast 块 */
			bool		clear_toast_afterwards;

			/* 删除 || 更新时有效 */
			ReorderBufferTupleBuf *oldtuple;
			/* 插入 || 更新时有效 */
			ReorderBufferTupleBuf *newtuple;
		}			tp;

		/*
		 * 为表示要截断的一组关系的 REORDER_BUFFER_CHANGE_TRUNCATE 截断数据。
		 */
		struct
		{
			Size		nrelids;
			bool		cascade;
			bool		restart_seqs;
			Oid		   *relids;
		}			truncate;

		/* 带有任意数据的消息。 */
		struct
		{
			char	   *prefix;
			Size		message_size;
			char	   *message;
		}			msg;

		/* 新快照，当 action == *_INTERNAL_SNAPSHOT 时设定 */
		Snapshot	snapshot;

		/*
		 * 在更改事务的目录中为现有快照的新命令 ID。当 action == *_INTERNAL_COMMAND_ID 时设定。
		 */
		CommandId	command_id;

		/*
		 * 目录更改事务的新 cid 映射，当 action == *_INTERNAL_TUPLECID 时设定。
		 */
		struct
		{
			RelFileNode node;
			ItemPointerData tid;
			CommandId	cmin;
			CommandId	cmax;
			CommandId	combocid;
		}			tuplecid;

		/* 无效化。 */
		struct
		{
			uint32		ninvalidations; /* 消息数量 */
			SharedInvalidationMessage *invalidations;	/* 无效化消息 */
		}			inval;
	}			data;

	/*
	 * 在使用过程中，这是如何将更改链接到事务中的，
	 * 否则它就是预分配的列表。
	 */
	dlist_node	node;
} ReorderBufferChange;

/* ReorderBufferTXN txn_flags */
#define RBTXN_HAS_CATALOG_CHANGES 0x0001
#define RBTXN_IS_SUBXACT          0x0002
#define RBTXN_IS_SERIALIZED       0x0004
#define RBTXN_IS_SERIALIZED_CLEAR 0x0008
#define RBTXN_IS_STREAMED         0x0010
#define RBTXN_HAS_PARTIAL_CHANGE  0x0020
#define RBTXN_PREPARE             0x0040
#define RBTXN_SKIPPED_PREPARE	  0x0080

/* 该事务是否有目录更改？ */
#define rbtxn_has_catalog_changes(txn) \
( \
	 ((txn)->txn_flags & RBTXN_HAS_CATALOG_CHANGES) != 0 \
)

/* 该事务是否被称为子事务？ */
#define rbtxn_is_known_subxact(txn) \
( \
	((txn)->txn_flags & RBTXN_IS_SUBXACT) != 0 \
)

/* 该事务是否已溢出到磁盘？ */
#define rbtxn_is_serialized(txn) \
( \
	((txn)->txn_flags & RBTXN_IS_SERIALIZED) != 0 \
)

/* 该事务是否曾经溢出到磁盘？ */
#define rbtxn_is_serialized_clear(txn) \
( \
	((txn)->txn_flags & RBTXN_IS_SERIALIZED_CLEAR) != 0 \
)

/* 该事务是否包含部分更改？ */
#define rbtxn_has_partial_change(txn) \
( \
	((txn)->txn_flags & RBTXN_HAS_PARTIAL_CHANGE) != 0 \
)

/*
 * 该事务是否已流向下游？
 *
 * （由于各种原因，无法从 nentries 和 nentries_mem 推断出这一点。
 * 例如，所有更改可能都在子事务中，在这种情况下，对于顶级事务
 * 来说 nentries==0，这不会说明任何关于流动的信息。因此我们保留这个标志，
 * 但是仅适用于顶级事务。）
 */
#define rbtxn_is_streamed(txn) \
( \
	((txn)->txn_flags & RBTXN_IS_STREAMED) != 0 \
)

/* 该事务是否已准备？ */
#define rbtxn_prepared(txn) \
( \
	((txn)->txn_flags & RBTXN_PREPARE) != 0 \
)

/* 该事务的准备是否被跳过？ */
#define rbtxn_skip_prepared(txn) \
( \
	((txn)->txn_flags & RBTXN_SKIPPED_PREPARE) != 0 \
)

typedef struct ReorderBufferTXN
{
	/* 见上文 */
	bits32		txn_flags;

	/* 事务的事务 ID，可以是顶级或子 xid。 */
	TransactionId xid;

	/* 顶级事务的 xid（如果已知） */
	TransactionId toplevel_xid;

	/*
	 * 识别已准备事务所需的全局事务 ID。
	 */
	char	   *gid;

	/*
	 * 第一个携带数据的 WAL 记录的 LSN，知道这个 xid。允许这个记录不是带有
	 * 这个 xid 的第一个记录，如果之前的记录与逻辑解码无关。
	 */
	XLogRecPtr	first_lsn;

	/* ----
	 * 导致此事务被准备、提交或中止的记录的 LSN。可以是
	 * * 普通提交记录
	 * * 父事务的普通提交记录
	 * * 已准备的事务
	 * * 已准备的事务提交
	 * * 普通中止记录
	 * * 已准备的事务中止
	 *
	 * 当事务溢出到磁盘时，这也可以设置为早于事务结束的值；
	 * 特别是设置为到目前为止写入磁盘的最新更改的 LSN。
	 * ----
	 */
	XLogRecPtr	final_lsn;

	/*
	 * 指向提交记录结尾 + 1 的 LSN。
	 */
	XLogRecPtr	end_lsn;

	/* 该子事务的顶级事务（对于顶级事务为 NULL）。 */
	struct ReorderBufferTXN *toptxn;

	/*
	 * 快照信息所在的最后一个 lsn 的 LSN，便于我们从那里重新启动解码
	 * 并从 WAL 完全恢复该事务。
	 */
	XLogRecPtr	restart_decoding_lsn;

	/* 导致此事务的更改来源 */
	RepOriginId origin_id;
	XLogRecPtr	origin_lsn;

	/*
	 * 提交或准备时间，仅在读取实际的提交或准备记录时已知。
	 */
	union
	{
		TimestampTz commit_time;
		TimestampTz prepare_time;
	}			xact_time;

	/*
	 * 基础快照用于解码所有更改，直到此事务修改目录或另一个修改目录的事务提交。
	 */
	Snapshot	base_snapshot;
	XLogRecPtr	base_snapshot_lsn;
	dlist_node	base_snapshot_node; /* 在 txns_by_base_snapshot_lsn 中链接 */

	/*
	 * 来自上一个流式运行的快照/CID。仅对已流式的事务有效（否则为 NULL/InvalidCommandId）。
	 */
	Snapshot	snapshot_now;
	CommandId	command_id;

	/*
	 * 在这个事务中我们有多少个 ReorderBufferChange。
	 *
	 * 子事务中的更改*不*包含在内，而是单独跟踪。
	 */
	uint64		nentries;

	/*
	 * 上述条目中有多少存储在内存中，而不是溢出到磁盘。
	 */
	uint64		nentries_mem;

	/*
	 * ReorderBufferChange 结构的列表，包括新的快照、新的 CommandIds 和命令失效消息。
	 */
	dlist_head	changes;

	/*
	 * (relation, ctid) => (cmin, cmax) 的映射列表，用于目录元组。
	 * 这些总是分配给顶层事务。（跟踪入口数量以创建合适大小的哈希）
	 */
	dlist_head	tuplecids;
	uint64		ntuplecids;

	/*
	 * 按需构建的哈希，以查找上述值。
	 */
	HTAB	   *tuplecid_hash;

	/*
	 * 包含（可能部分）toast 条目的哈希。如果没有找到当前更改的 toast 元组，则为 NULL。
	 */
	HTAB	   *toast_hash;

	/*
	 * 非层次化的子事务列表，这些子事务*不*被中止。仅在顶层事务中使用。
	 */
	dlist_head	subtxns;
	uint32		nsubtxns;

	/*
	 * 存储的缓存失效。这不是链表，因为我们一次获取所有失效信息。
	 */
	uint32		ninvalidations;
	SharedInvalidationMessage *invalidations;

	/* ---
	 * 在三个列表中的位置：
	 * * 如果我们*已知*为子事务，则在子事务列表中
	 * * 在顶层事务列表中（可以是尚未知的子事务）
	 * * 预分配 ReorderBufferTXNs 的列表（如果未使用）
	 * ---
	 */
	dlist_node	node;

	/*
	 * 此事务的大小（目前在内存中的更改，以字节为单位）。
	 */
	Size		size;

	/* 包含子事务的顶层事务的大小。 */
	Size		total_size;

	/* 如果我们检测到并发中止，则忽略未来的更改。 */
	bool		concurrent_abort;

	/*
	 * 输出插件的私有数据指针。
	 */
	void	   *output_plugin_private;
} ReorderBufferTXN;

/* 以便定义在结构 ReorderBuffer 内部使用的回调 */
typedef struct ReorderBuffer ReorderBuffer;

/* 更改回调签名 */
typedef void (*ReorderBufferApplyChangeCB) (ReorderBuffer *rb,
											ReorderBufferTXN *txn,
											Relation relation,
											ReorderBufferChange *change);

/* 截断回调签名 */
typedef void (*ReorderBufferApplyTruncateCB) (ReorderBuffer *rb,
											  ReorderBufferTXN *txn,
											  int nrelations,
											  Relation relations[],
											  ReorderBufferChange *change);

/* 开始回调签名 */
typedef void (*ReorderBufferBeginCB) (ReorderBuffer *rb,
									  ReorderBufferTXN *txn);

/* 提交回调签名 */
typedef void (*ReorderBufferCommitCB) (ReorderBuffer *rb,
									   ReorderBufferTXN *txn,
									   XLogRecPtr commit_lsn);

/* 消息回调签名 */
typedef void (*ReorderBufferMessageCB) (ReorderBuffer *rb,
										ReorderBufferTXN *txn,
										XLogRecPtr message_lsn,
										bool transactional,
										const char *prefix, Size sz,
										const char *message);

/* 开始准备回调签名 */
typedef void (*ReorderBufferBeginPrepareCB) (ReorderBuffer *rb,
											 ReorderBufferTXN *txn);

/* 准备回调签名 */
typedef void (*ReorderBufferPrepareCB) (ReorderBuffer *rb,
										ReorderBufferTXN *txn,
										XLogRecPtr prepare_lsn);

/* 提交准备好的回调签名 */
typedef void (*ReorderBufferCommitPreparedCB) (ReorderBuffer *rb,
											   ReorderBufferTXN *txn,
											   XLogRecPtr commit_lsn);

/* 回滚准备好的回调签名 */
typedef void (*ReorderBufferRollbackPreparedCB) (ReorderBuffer *rb,
												 ReorderBufferTXN *txn,
												 XLogRecPtr prepare_end_lsn,
												 TimestampTz prepare_time);

/* 启动流式事务回调签名 */
typedef void (*ReorderBufferStreamStartCB) (
											ReorderBuffer *rb,
											ReorderBufferTXN *txn,
											XLogRecPtr first_lsn);

/* 停止流式事务回调签名 */
typedef void (*ReorderBufferStreamStopCB) (
										   ReorderBuffer *rb,
										   ReorderBufferTXN *txn,
										   XLogRecPtr last_lsn);

/* 丢弃流式事务回调签名 */
typedef void (*ReorderBufferStreamAbortCB) (
											ReorderBuffer *rb,
											ReorderBufferTXN *txn,
											XLogRecPtr abort_lsn);

/* 准备流式事务回调签名 */
typedef void (*ReorderBufferStreamPrepareCB) (
											  ReorderBuffer *rb,
											  ReorderBufferTXN *txn,
											  XLogRecPtr prepare_lsn);

/* 提交流式事务回调签名 */
typedef void (*ReorderBufferStreamCommitCB) (
											 ReorderBuffer *rb,
											 ReorderBufferTXN *txn,
											 XLogRecPtr commit_lsn);

/* 流更改回调签名 */
typedef void (*ReorderBufferStreamChangeCB) (
											 ReorderBuffer *rb,
											 ReorderBufferTXN *txn,
											 Relation relation,
											 ReorderBufferChange *change);

/* 流消息回调签名 */
typedef void (*ReorderBufferStreamMessageCB) (
											  ReorderBuffer *rb,
											  ReorderBufferTXN *txn,
											  XLogRecPtr message_lsn,
											  bool transactional,
											  const char *prefix, Size sz,
											  const char *message);

/* 流截断回调签名 */
typedef void (*ReorderBufferStreamTruncateCB) (
											   ReorderBuffer *rb,
											   ReorderBufferTXN *txn,
											   int nrelations,
											   Relation relations[],
											   ReorderBufferChange *change);

struct ReorderBuffer
{
	/*
	 * xid => ReorderBufferTXN 查找表
	 */
	HTAB	   *by_txn;

	/*
	 * 可以是顶层事务的事务，按第一个
	 * 记录的 LSN 排序，该记录携带此 xid。
	 */
	dlist_head	toplevel_by_lsn;

	/*
	 * 具有基础快照的事务和子事务，按
	 * 造成我们首次获得基础快照的记录的 LSN 排序。
	 * 这与 toplevel_by_lsn 不同，因为我们仅在第一个
	 * 逻辑解码相关记录（例如堆写入）上设置基础
	 * 快照，而初始 LSN 可能由其他操作设置。
	 */
	dlist_head	txns_by_base_snapshot_lsn;

	/*
	 * 一个条目大小的缓存，适用于 by_txn。非常频繁地
	 * 相同的事务被查找多次。
	 */
	TransactionId by_txn_last_xid;
	ReorderBufferTXN *by_txn_last_txn;

	/*
	 * 在事务提交时调用的回调。
	 */
	ReorderBufferBeginCB begin;
	ReorderBufferApplyChangeCB apply_change;
	ReorderBufferApplyTruncateCB apply_truncate;
	ReorderBufferCommitCB commit;
	ReorderBufferMessageCB message;

	/*
	 * 在准备阶段流式传输事务时调用的回调。
	 */
	ReorderBufferBeginCB begin_prepare;
	ReorderBufferPrepareCB prepare;
	ReorderBufferCommitPreparedCB commit_prepared;
	ReorderBufferRollbackPreparedCB rollback_prepared;

	/*
	 * 在流式传输事务时调用的回调。
	 */
	ReorderBufferStreamStartCB stream_start;
	ReorderBufferStreamStopCB stream_stop;
	ReorderBufferStreamAbortCB stream_abort;
	ReorderBufferStreamPrepareCB stream_prepare;
	ReorderBufferStreamCommitCB stream_commit;
	ReorderBufferStreamChangeCB stream_change;
	ReorderBufferStreamMessageCB stream_message;
	ReorderBufferStreamTruncateCB stream_truncate;

	/*
	 * 将被原封不动地传递给回调的指针。
	 */
	void	   *private_data;

	/*
	 * 保存的输出插件选项
	 */
	bool		output_rewrites;

	/*
	 * 私有内存上下文。
	 */
	MemoryContext context;

	/*
	 * 特定类型对象的内存上下文
	 */
	MemoryContext change_context;
	MemoryContext txn_context;
	MemoryContext tup_context;

	XLogRecPtr	current_restart_decoding_lsn;

	/* 磁盘<->内存转换的缓冲区 */
	char	   *outbuf;
	Size		outbufsize;

	/* 内存核算 */
	Size		size;

	/*
	 * 写入磁盘的事务统计信息。
	 *
	 * 单个事务可能会重复写入磁盘，这就是为什么我们保持
	 * 两个不同计数器的原因。对于写入磁盘，事务计数器包含
	 * 顶层事务和子事务。
	 */
	int64		spillTxns;		/* 写入磁盘的事务数量 */
	int64		spillCount;		/* 写入磁盘调用计数器 */
	int64		spillBytes;		/* 写入磁盘的数据量 */

	/* 向解码输出插件流式传输的事务统计信息 */
	int64		streamTxns;		/* 流式传输的事务数量 */
	int64		streamCount;	/* 流式传输调用计数器 */
	int64		streamBytes;	/* 解码的数据量 */

	/*
	 * 发送到解码输出的所有事务的统计信息
	 * 插件
	 */
	int64		totalTxns;		/* 发送的事务总数 */
	int64		totalBytes;		/* 解码的数据总量 */
};


extern ReorderBuffer *ReorderBufferAllocate(void);
extern void ReorderBufferFree(ReorderBuffer *);

extern ReorderBufferTupleBuf *ReorderBufferGetTupleBuf(ReorderBuffer *, Size tuple_len);
extern void ReorderBufferReturnTupleBuf(ReorderBuffer *, ReorderBufferTupleBuf *tuple);
extern ReorderBufferChange *ReorderBufferGetChange(ReorderBuffer *);
extern void ReorderBufferReturnChange(ReorderBuffer *, ReorderBufferChange *, bool);

extern Oid *ReorderBufferGetRelids(ReorderBuffer *, int nrelids);
extern void ReorderBufferReturnRelids(ReorderBuffer *, Oid *relids);

extern void ReorderBufferQueueChange(ReorderBuffer *, TransactionId,
									 XLogRecPtr lsn, ReorderBufferChange *,
									 bool toast_insert);
extern void ReorderBufferQueueMessage(ReorderBuffer *, TransactionId, Snapshot snapshot, XLogRecPtr lsn,
									  bool transactional, const char *prefix,
									  Size message_size, const char *message);
extern void ReorderBufferCommit(ReorderBuffer *, TransactionId,
								XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
								TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn);
extern void ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,
										XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
										XLogRecPtr two_phase_at,
										TimestampTz commit_time,
										RepOriginId origin_id, XLogRecPtr origin_lsn,
										char *gid, bool is_commit);
extern void ReorderBufferAssignChild(ReorderBuffer *, TransactionId, TransactionId, XLogRecPtr commit_lsn);
extern void ReorderBufferCommitChild(ReorderBuffer *, TransactionId, TransactionId,
									 XLogRecPtr commit_lsn, XLogRecPtr end_lsn);
extern void ReorderBufferAbort(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
extern void ReorderBufferAbortOld(ReorderBuffer *, TransactionId xid);
extern void ReorderBufferForget(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
extern void ReorderBufferInvalidate(ReorderBuffer *, TransactionId, XLogRecPtr lsn);

extern void ReorderBufferSetBaseSnapshot(ReorderBuffer *, TransactionId, XLogRecPtr lsn, struct SnapshotData *snap);
extern void ReorderBufferAddSnapshot(ReorderBuffer *, TransactionId, XLogRecPtr lsn, struct SnapshotData *snap);
extern void ReorderBufferAddNewCommandId(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
										 CommandId cid);
extern void ReorderBufferAddNewTupleCids(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
										 RelFileNode node, ItemPointerData pt,
										 CommandId cmin, CommandId cmax, CommandId combocid);
extern void ReorderBufferAddInvalidations(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
										  Size nmsgs, SharedInvalidationMessage *msgs);
extern void ReorderBufferImmediateInvalidation(ReorderBuffer *, uint32 ninvalidations,
											   SharedInvalidationMessage *invalidations);
extern void ReorderBufferProcessXid(ReorderBuffer *, TransactionId xid, XLogRecPtr lsn);

extern void ReorderBufferXidSetCatalogChanges(ReorderBuffer *, TransactionId xid, XLogRecPtr lsn);
extern bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *, TransactionId xid);
extern bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *, TransactionId xid);

extern bool ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid,
											 XLogRecPtr prepare_lsn, XLogRecPtr end_lsn,
											 TimestampTz prepare_time,
											 RepOriginId origin_id, XLogRecPtr origin_lsn);
extern void ReorderBufferSkipPrepare(ReorderBuffer *rb, TransactionId xid);
extern void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid, char *gid);
extern ReorderBufferTXN *ReorderBufferGetOldestTXN(ReorderBuffer *);
extern TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *rb);

extern void ReorderBufferSetRestartPoint(ReorderBuffer *, XLogRecPtr ptr);

extern void StartupReorderBuffer(void);

#endif
